package com.simple.mqtt.common.task;

import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.BiConsumer;

/**
 * MQTT message retransmission processor
 * Manages scheduled retransmission mechanism for MQTT messages
 * Reference from netty-mqtt-client implementation
 *
 * @param <T> message type generic, must extend MqttMessage
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public final class RetryProcessor<T extends MqttMessage> {

    private AckTimerTask ackTimerTask;
    private BiConsumer<MqttFixedHeader, T> handler;
    /** Original message object */
    private T originalMessage;

    /**
     * Start retransmission mechanism
     * @param taskService scheduled task service
     * @throws NullPointerException when handler or taskService is null
     */
    public void start(ScheduledExecutorService taskService) {
        Objects.requireNonNull(this.handler, "RetryProcessor handler is null.");
        this.startTimer(Objects.requireNonNull(taskService, "RetryProcessor taskService is null."));
    }

    /**
     * Start retransmission timer
     * @param taskService scheduled task service
     */
    private void startTimer(ScheduledExecutorService taskService) {
        // Create scheduled retransmission task
        this.ackTimerTask = new AckTimerTask(taskService, () -> {
            // Create new fixed header, set DUP flag to true indicating this is a retransmission message
            MqttFixedHeader fixedHeader = new MqttFixedHeader(
                    this.originalMessage.fixedHeader().messageType(),
                    true, // Retry flag set to true
                    this.originalMessage.fixedHeader().qosLevel(),
                    this.originalMessage.fixedHeader().isRetain(),
                    this.originalMessage.fixedHeader().remainingLength()
            );
            // Execute retransmission processing
            handler.accept(fixedHeader, originalMessage);
        }, 5, 10); // Initial delay 5 seconds, interval 10 seconds

        // Start scheduled task
        this.ackTimerTask.start();
    }

    /**
     * Stop retransmission mechanism
     * Cancel scheduled retransmission task
     */
    public void stop() {
        if (this.ackTimerTask != null) {
            // Cancel scheduled task
            this.ackTimerTask.cancel();
        }
    }
}